package com.muyu.vehicle.hbase.service;

import com.muyu.vehicle.entity.RowData;
import com.muyu.vehicle.hbase.config.HbaseConfig;
import com.muyu.vehicle.hbase.instance.HbaseInstance;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author 牧鱼
 * @Classname HbaseDataService
 * @Description
 * @Date 2021/8/26
 */
@Component
public class HbaseDataService {

    private Connection conn = null;

    public HbaseDataService(HbaseInstance hbaseInstance) {
        conn = hbaseInstance.getConn();
    }

    /**
     * 获取行键
     * @param prefixString
     * @param suffixString
     * @param options
     * @return
     */
    public String getRowKey(String prefixString, String suffixString, Object... options) {
        if (prefixString.length() > HbaseConfig.ROW_KEY_PRE_LENGTH || suffixString.length() > HbaseConfig.ROW_KEY_SUF_LENGTH) {
            return null;
        }
        StringBuilder preStringBuilder = new StringBuilder();
        preStringBuilder.append(prefixString);

        for (int i = 0; i < (HbaseConfig.ROW_KEY_PRE_LENGTH - preStringBuilder.length()); i++) {
            preStringBuilder.append("0");
        }

        StringBuilder sufStringBuilder = new StringBuilder();
        sufStringBuilder.append(suffixString);

        for (int i = 0; i < (HbaseConfig.ROW_KEY_SUF_LENGTH - sufStringBuilder.length()); i++) {
            sufStringBuilder.append("0");
        }

        StringBuilder optBuilder = new StringBuilder();

        for (Object option : options) {
            optBuilder.append(option);
        }

        if (optBuilder.length() > HbaseConfig.ROW_KEY_OPTIONS_LENGTH) {
            return null;
        }

        StringBuilder optStringBuilder = new StringBuilder();
        for (int i = 0; i < (HbaseConfig.ROW_KEY_OPTIONS_LENGTH - optBuilder.length()); i++) {
            optStringBuilder.append("0");
        }

        optStringBuilder.append(optBuilder);

        return preStringBuilder.append("|").append(optStringBuilder).append("|").append(sufStringBuilder).toString();
    }


    /**
     * 插入一条记录
     *
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     * @param value
     * @throws IOException
     */
    public void insertOne(String tableName, String rowKey, String columnFamily, String column, String value) throws IOException {
        Put put = new Put(Bytes.toBytes(rowKey));
        //下面三个分别为，列族，列名，列值
        put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
        //得到 table
        Table table = conn.getTable(TableName.valueOf(tableName));
        //执行插入
        table.put(put);
    }

    /**
     * 多行多列多值-单列簇
     *
     * @param tableName
     * @param columnFamily
     * @param mapList
     * @throws IOException
     */
    public void insertAll(String tableName, String columnFamily, List<Map<String, String>> mapList) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        List<Put> puts = new ArrayList<>();

        for (Map<String, String> map : mapList) {
            Put put = new Put(Bytes.toBytes(map.get("rowKey")));
            put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(map.get("column")), Bytes.toBytes(map.get("value")));
            puts.add(put);
        }
        table.put(puts);
    }

    /**
     * 向HBase表中批量插入数据
     * 该方法主要用于在给定的HBase表中，针对特定行键(rowKey)和列族(columnFamily),
     * 批量插入由键值对表示的数据项
     *
     * @param tableName 表名称
     * @param rowKey 行键，标识HBase表中的行
     * @param columnFamily 列族，具有相同列族的列的集合
     * @param mapData 包含要插入的数据的Map，键将用作列名，值将用作单元格值
     */
    public void insertAll(String tableName, String rowKey, String columnFamily, Map<String, String> mapData) throws IOException {
        // 为mapData添加纬度和经度信息，这些是固定的数据项
        mapData.put("latitude", "116.388300");
        mapData.put("longitude", "21.928900");
        // 使用try-with-resources确保Table资源在使用后自动关闭
        try (Table table = conn.getTable(TableName.valueOf(tableName))) {
            // 创建一个Put对象的列表，用于批量插入数据
            List<Put> puts = new ArrayList<>();
            // 遍历mapData中的每一个键值对，创建并填充Put对象
            for (Map.Entry<String, String> entry : mapData.entrySet()) {
                // 创建一个新的Put对象，并指定行键
                Put put = new Put(Bytes.toBytes(rowKey));
                // 添加列，列名由mapData的键转换而来，值也是
                put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(entry.getKey()), Bytes.toBytes(entry.getValue()));
                // 将Put对象添加到列表中
                puts.add(put);
            }
            // 如果有数据需要插入，则批量写入HBase表中
            if (!puts.isEmpty()) {
                table.put(puts);
            }
        } catch (TableNotFoundException e) {
            // 如果表不存在，抛出异常并包含错误信息和原因
            throw new IOException("Table not found: " + tableName, e);
        } catch (Exception e) {
            // 捕获其他所有异常，并转换为IOException，包含通用的错误信息和原因
            throw new IOException("Failed to insert data into HBase", e);
        }
    }

    /**
     * 更新数据
     *
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     * @param value
     * @throws IOException
     */
    public void update(String tableName, String rowKey, String columnFamily, String column, String value) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
        table.put(put);
        // table.close();
    }

    /**
     * 删除单行单列
     *
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     * @throws IOException
     */
    public void delete(String tableName, String rowKey, String columnFamily, String column) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        delete.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
        table.delete(delete);
    }

    /**
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param columnList
     * @throws IOException
     */
    public void delete(String tableName, String rowKey, String columnFamily, String... columnList) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        for (String column : columnList) {
            delete.addColumns(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
        }
        table.delete(delete);
    }

    /**
     * 删除单行单列簇
     *
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @throws IOException
     */
    public void delete(String tableName, String rowKey, String columnFamily) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        delete.addFamily(Bytes.toBytes(columnFamily));
        table.delete(delete);
    }

    /**
     * 删除单行
     *
     * @param tableName
     * @param rowKey
     * @throws IOException
     */
    public void delete(String tableName, String rowKey) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        table.delete(delete);
    }

    /**
     * 删除多行
     *
     * @param tableName
     * @param rowKeyList
     * @throws IOException
     */
    public void delete(String tableName, String... rowKeyList) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        ArrayList<Delete> deleteList = new ArrayList<>();
        for (String rowKey : rowKeyList) {
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            deleteList.add(delete);
        }
        table.delete(deleteList);
    }

    /**
     * 查询表
     *
     * @param tableName
     * @param rowKey
     * @return
     * @throws IOException
     */
    public Result select(String tableName, String rowKey) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        Get get = new Get(Bytes.toBytes(rowKey));
        Result result = table.get(get);
        return result;

    }

    /**
     * 全表扫描
     *
     * @return
     * @throws IOException
     */
    public ResultScanner scan(String tableName) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        ResultScanner resultScanner = table.getScanner(scan);
        return resultScanner;
    }

    /**
     * 全表扫描-列簇
     *
     * @param tableName
     * @param columnFamily
     * @return
     * @throws IOException
     */
    public ResultScanner scan(String tableName, String columnFamily) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.addFamily(Bytes.toBytes(columnFamily));
        ResultScanner scanner = table.getScanner(scan);

        return scanner;

    }

    /**
     * 全表扫描-列
     *
     * @param tableName
     * @param columnFamily
     * @param column
     * @return
     * @throws IOException
     */
    public ResultScanner scan(String tableName, String columnFamily, String column) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
        ResultScanner scanner = table.getScanner(scan);

        return scanner;
    }

    /**
     * 全表扫描-过滤器
     *
     * @param tableName
     * @param filters
     * @return
     * @throws IOException
     */
    public ResultScanner scan(String tableName, Filter... filters) throws IOException {

        Table table = conn.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        for (Filter filter : filters) {
            scan.setFilter(filter);
        }
        ResultScanner scanner = table.getScanner(scan);

        return scanner;
    }

    /**
     * 分页全表扫描-过滤器
     *
     * @param tableName
     * @param filter
     * @return
     * @throws IOException
     */
    public ResultScanner scan(String tableName, Filter filter, String startRowKey) throws IOException {

        Table table = conn.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.setFilter(filter);
        scan.withStartRow(Bytes.toBytes(startRowKey));
        ResultScanner scanner = table.getScanner(scan);

        return scanner;
    }

    /**
     * 获取分页过滤器
     *
     * @param size
     * @return
     */
    public Filter pagetFilter(long size) {

        return new PageFilter(size);

    }

    /**
     * singleColumnValueFilter
     *
     * @param columnFamily
     * @param column
     * @param compareOperator
     * @param value
     * @return
     */
    public Filter singleColumnValueFilter(String columnFamily, String column, CompareOperator compareOperator, String value) {
        return new SingleColumnValueFilter(Bytes.toBytes(columnFamily), Bytes.toBytes(column), compareOperator, Bytes.toBytes(value));
    }

    /**
     * rowFilter
     *
     * @param compareOperator
     * @param rowComparator
     * @return
     */
    public Filter rowFilter(CompareOperator compareOperator, ByteArrayComparable rowComparator) {
        return new RowFilter(compareOperator, rowComparator);
    }

    /**
     * columnPrefixFilter
     *
     * @param prefix
     * @return
     */
    public Filter columnPrefixFilter(String prefix) {
        return new ColumnPrefixFilter(Bytes.toBytes(prefix));
    }

    /**
     * 过滤器集合
     *
     * @param filterList
     * @return
     */
    public FilterList filterListPassAll(Filter... filterList) {
        FilterList list = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        for (Filter filter : filterList) {
            list.addFilter(filter);
        }
        return list;
    }

    /**
     * 过滤器集合
     *
     * @param
     * @return
     */
    public FilterList filterListPassOne(Filter... filterList) {
        FilterList list = new FilterList(FilterList.Operator.MUST_PASS_ONE);
        for (Filter filter : filterList) {
            list.addFilter(filter);
        }
        return list;
    }


    /**
     * 关闭连接
     *
     * @throws IOException
     */
    public void close() {
        try {
            conn.close();
        } catch (IOException e) {
            conn = null;
        } finally {
            conn = null;
        }
    }

    public List<Map<String, String>> getAllData(String tableName) throws IOException {
        List<Map<String, String>> resultList = new ArrayList<>();
        Table table = conn.getTable(TableName.valueOf(tableName));
        try (ResultScanner scanner = table.getScanner(new Scan())) {
            for (Result result : scanner) {
                Map<String, String> row = new HashMap<>();
                byte[] rowKey = result.getRow();
                row.put("rowKey", Bytes.toString(rowKey));

                for (Cell cell : result.rawCells()) {
                    byte[] family = CellUtil.cloneFamily(cell);
                    byte[] qualifier = CellUtil.cloneQualifier(cell);
                    byte[] value = CellUtil.cloneValue(cell);

                    String familyStr = Bytes.toString(family);
                    String qualifierStr = Bytes.toString(qualifier);
                    String valueStr = Bytes.toString(value);

                    row.put("columnFamily", familyStr);
                    row.put("column", qualifierStr);
                    row.put("value", valueStr);
                }
                resultList.add(row);
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return resultList;
    }


    public long countRowKeys(String tableName) throws IOException {
        try (Table table = conn.getTable(TableName.valueOf(tableName))) {
            Scan scan = new Scan();
            ResultScanner scanner = table.getScanner(scan);
            List<Result> results = new ArrayList<>();
            for (Result result : scanner) {
                results.add(result);
            }
            return results.size();
        }
    }

    public List<RowData> getData(String tableName) throws IOException {
        // 存储查询结果的列表
        List<RowData> resultList = new ArrayList<>();
        // 根据表名获取Table对象
        Table table = conn.getTable(TableName.valueOf(tableName));
        // 创建Scanner对象，用于扫描表中所有数据
        ResultScanner scanner = table.getScanner(new Scan());
        // 遍历查询结果
        for (Result result : scanner) {
            // 处理每一行数据
            String rowkey1 = Bytes.toString(result.getRow());
            RowData rowData = new RowData(rowkey1);
            Cell[] cells = result.rawCells();
            Map<String, Object> map = new HashMap<>();
            // 遍历每一行的单元格，提取列名和值
            for (Cell cell : cells) {
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                System.out.println(rowkey + "----" + cf + "--" + column + "---" + value);
                // 将列名和值添加到rowData对象
                rowData.addColumn(column, value);
                map.put(column,value);
            }
            // 将当前行的数据信息添加到rowData对象
            rowData.setMap(map);
            // 将rowData对象添加到结果列表
            resultList.add(rowData);
        }
        // 关闭Scanner和Table对象，释放资源
        scanner.close();
        table.close();
        // 返回查询结果列表
        return resultList;
    }
}
